package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Locale;

/**
 * 个股秒级窗口函数处理对象
 */
public class StockSecWindowFunction extends RichWindowFunction<CleanBean, StockBean, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<CleanBean> input, Collector<StockBean> out) throws Exception {

        /**
         * 开发步骤：
         * 1.记录最新个股
         * 2.格式化日期
         * 3.封装输出数据
         */
        //1.记录最新个股
        CleanBean newCleanBean = null;
        for (CleanBean cleanBean : input) {
            if(newCleanBean == null){
                newCleanBean = cleanBean;
            }
            if(newCleanBean.getEventTime() < cleanBean.getEventTime()){
                newCleanBean = cleanBean;
            }
        }

        /**
         * 加盐/随机数+ 唯一标识 + 时间戳 (此方法不满足连续查询的业务需求)
         * 唯一标识(证券代码) + 时间戳 （本项目用此方法）
         * startkey = 300059+ 20240426093000  =30005920240426093000 (起始时间)
         * endkey = 300059+ 20240426093059  = 30005920240426093059 (结束时间)
         *
         * 假设查询601375,从早上9:30到11:00的连续秒级行情数据
         * 开始时间: 60137520230729093000
         * 结束时间: 60137520230729110000
         * Scan scan = new Scan()
         * scan.setStartRow()
         * scan.setEndRow()
         */
        //2.格式化日期
        Long tradeTime = DateUtil.longTimeTransfer(newCleanBean.getEventTime(), Constant.format_YYYYMMDDHHMMSS);
        String tradeDate = DateUtil.longTimeToString(newCleanBean.getEventTime(), Constant.format_yyyy_mm_dd);

        //3.封装输出数据
        out.collect(new StockBean(
                tradeTime,
                newCleanBean.getSecCode(),
                newCleanBean.getSecName(),
                newCleanBean.getPreClosePrice(),
                newCleanBean.getOpenPrice(),
                newCleanBean.getHighPrice(),
                newCleanBean.getLowPrice(),
                newCleanBean.getTradePrice(),
                0l,0l,
                newCleanBean.getTradeVolumn(),
                newCleanBean.getTradeAmt(),
                tradeDate,
                newCleanBean.getEventTime(),
                newCleanBean.getSource()
        ));


    }

}
